%%% Copyright 2010 Google Inc.
%%%
%%% Licensed under the Apache License, Version 2.0 (the "License");
%%% you may not use this file except in compliance with the License.
%%% You may obtain a copy of the License at
%%%
%%%     http://www.apache.org/licenses/LICENSE-2.0
%%%
%%% Unless required by applicable law or agreed to in writing, software
%%% distributed under the License is distributed on an "AS IS" BASIS,
%%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%%% See the License for the specific language governing permissions and
%%% limitations under the License.

%%% @copyright 2010 Google Inc. All Rights Reserved.
%%% @author <jacoby@google.com> Jacoby Thwaites [http://sites/madcloud]
%%% @version {@version}
%%% @doc
%%% Each node has a node_srv. It is responsible for starting up the Mnesia
%%% tables for the node, creating them if necessary either from scratch
%%% in a 'local' startup, or from any existing node if specified in
%%% start_link/1.
%%%
%%% Each node_srv maintains a latency table to other nodes. This allows
%%% fieldset_srv emissions to be sent to the 'closest' view_fsm for
%%% automatic latency optimization.
%%%
%%% @headerfile "../include/txm.hrl"

-module(node_srv).

-behaviour(gen_server).

-include("txm.hrl").

%% A list of all Mnesia tables in the system.
-define(ALL_TABLES, [activation | ?TABLES]).

%% gen_server call tag used to check this node_srv is up.
-define(READY, ready).

%% The latency ETS table is updated by this node_srv, and read in-process
%% by other processes, notably fieldset_srv when emitting input to ops.
-define(LATENCY, latency).
-define(UNFEASIBLY_LARGE_ROUNDTRIP_MICROS, (60 * 1000 * 1000)).
-record(?LATENCY, {node = ?UNDEFINED :: atom(),
                   roundtrip = 0     :: integer()}).

%% The update server call tag and frequency for the latency table.
-define(LATENCY_REFRESH, latency_refresh).
-define(LATENCY_REFRESH_MILLIS, (60 * 1000)).

%% A single node server per node, one is global txm_master.
-export([start_link/1, stop/0]).
-export([latency_refresh/0]).
-export([ready/0, closest/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, code_change/3]).
-export([terminate/2]).

%% ----------------------------------------------------------------------------
%% User/client functions.
%% ----------------------------------------------------------------------------

%% @doc Starts the server for this node. Mode can be 'local' or a list of 1
%% or more already started nodes in the cluster. The first of these that
%% is up is used as the source for the new replica.
start_link(?STARTUP_LOCAL) ->
  gen_server:start_link({local, ?MODULE}, ?MODULE, ?STARTUP_LOCAL, []);

%% #2
start_link([Node | _Rest] = NodeNames) when is_atom(Node) ->
  gen_server:start_link({local, ?MODULE}, ?MODULE, NodeNames, []).

%% @doc Stops the server.
stop() ->
  gen_server:call(?MODULE, ?STOP).

%% @doc Returns ok when this server is up. Used by remote nodes to determine
%% if this node is up.
ready() ->
  gen_server:call(?MODULE, ?READY).

%% @doc Causes the node_srv to refresh its latency table.
latency_refresh() ->
  gen_server:cast(?MODULE, ?LATENCY_REFRESH).

%% @doc Returns the 'closest' of the supplied non-empty list of nodes, the
%% node with the lowest roundtrip time recorded in the latency table. This
%% is run in the caller's process. This always returns one of the nodes in
%% the supplied `Nodes', even if the table is not yet populated with its
%% roundtrip.
closest([First|_] = Nodes) ->

  case ets:info(?LATENCY) of
    ?UNDEFINED ->
      ?WARNING([closest, {no_table, ?LATENCY}]),
      First;
    _TableExists ->
      % We invent a correct record with an unfeasibly large roundtrip
      % to compare the others against.
      FirstRecord = #?LATENCY{node = First,
                              roundtrip = ?UNFEASIBLY_LARGE_ROUNDTRIP_MICROS},

      % Scan the nodes and return the one whose recorded roundtrip is lowest.
      BestRecord = lists:foldl(
                     fun(Node, #?LATENCY{roundtrip = BestSoFar} = Acc) ->
                          case ets:lookup(?LATENCY, Node) of
                            [#?LATENCY{roundtrip = Roundtrip} = Next]
                              when Roundtrip < BestSoFar ->
                              Next;
                            _Otherwise ->
                              Acc
                            end
                     end, FirstRecord, Nodes),
      BestRecord#?LATENCY.node
  end.

%% ----------------------------------------------------------------------------
%% gen_server callbacks.
%% ----------------------------------------------------------------------------

%% @doc Init callback returns initial state.
init(local) ->
  ?INFO([init, {node, local}]),
  process_flag(trap_exit, true),
  local_tables(),
  init_latency(),
  ?OK_RETURN([]);

%% #2
init(NodeNames) when is_list(NodeNames) ->
  ?INFO([init, {using, NodeNames}]),
  process_flag(trap_exit, true),
  replica_tables(NodeNames),
  init_latency(),
  ?OK_RETURN([]).

%% No action on terminate.
terminate(Reason, _State) ->
  ?INFO([{terminate, Reason}]),
  ?OK_RETURN.

%% Handles synchronous client calls.
%% #1
handle_call(?STOP, _From, State) ->
  {?STOP, normal, ?OK_RETURN, State};

%% #2
handle_call(?READY, _From, State) ->
  {reply, ?OK_RETURN, State}.

%% Handles client asynchronous calls.
handle_cast(?LATENCY_REFRESH, State) ->
  do_latency_refresh(),
  {noreply, State}.

%% Handles trapped exit message.
handle_info({'EXIT', Pid, Reason}, State) ->
  ?INFO([exit, {pid, Pid}, {reason, Reason}]),
  {?STOP, Reason, State}.

%% ----------------------------------------------------------------------------
%% Supporting functions.
%% ----------------------------------------------------------------------------

%% @doc Initializes the latency table and the timer that refreshes it. The
%% table is read-only by all processes and named because there is only one
%% per node. Returns the timer reference.
%% @spec init_latency() -> term()
init_latency() ->
  ?INFO([create_table, {ets, ?LATENCY}]),
  ets:new(?LATENCY, [named_table, set, protected, {keypos, #?LATENCY.node}]),
  latency_refresh(),
  ?OK_RETURN(TimerRef) =
    timer:apply_interval(?LATENCY_REFRESH_MILLIS, ?MODULE, latency_refresh, []),
  TimerRef.

%% @doc Refreshes the latency table by pinging all connected nodes including
%% this one and recording the roundtrip time. Each update is atomic, so it is
%% possible for a read to take place between updates, and even for a node to
%% be removed that is no longer connected.
%% Returns the number of connected nodes.
%% @spec do_latency_refresh() -> integer()
do_latency_refresh() ->
  Result = lists:foldl(fun(Node, Count) ->
                   case timer:tc(net_adm, ping, [Node]) of
                     {Micros, pong} ->
                       ets:insert(?LATENCY, #?LATENCY{node = Node,
                                                      roundtrip = Micros});
                     {_Micros, pang} ->
                       ?WARNING([latency_refresh, {cannot_reach, Node}]),
                       ets:delete(?LATENCY, Node)
                     end,
                   Count + 1
              end, 0, [node() | nodes()]),
  ?DEBUG([latency_refresh, {node_count, Result}]),
  Result.

%% @doc Checks whether a local copy of the Mnesia tables exists on this node
%% and creates them if not.
local_tables() ->
  LocalTables = mnesia:system_info(local_tables),
  case lists:subtract(?TABLES, LocalTables) of
    [] ->
      mnesia:wait_for_tables(?TABLES, ?OPEN_TIMEOUT),
      ?OK_RETURN;
    [_|_] ->
      prog:create_tables(),
      view_db:create_tables()
  end.

%% @doc Opens a replica of the tables which must be present on `ExistingNode'.
replica_tables(ExistingNodes) ->

  % Determine the first node that's up and running with a node_srv.
  Source = txm_util:first(
             fun(Node) ->
                  case rpc:call(Node, ?MODULE, ready, []) of
                    {badrpc, _Reason} ->
                      false;
                    ?OK_RETURN ->
                      Node
                    end
             end, ExistingNodes),

  ?OK_RETURN([Source]) = mnesia:change_config(extra_db_nodes, [Source]),
  lists:foreach(
    fun(Tab) ->
         Here = node(),
         mnesia:add_table_copy(Tab, Here, ram_copies)
    end, ?ALL_TABLES).

%% ----------------------------------------------------------------------------
%% Unimplemented gen_server callbacks.
%% ----------------------------------------------------------------------------

code_change(_, _, _) -> throw(?REASON_NOT_IMPLEMENTED).
